package writer

import (
	"fmt"

	"v8s-ssp-proxy/internal/configuration"
	"v8s/logger"

	"github.com/Shopify/sarama"

	"go.uber.org/zap"
)

type (
	Writer struct {
		log *zap.SugaredLogger

		producer sarama.SyncProducer
		topic    string
	}
)

func NewWriter(properties configuration.KafkaProperties) *Writer {
	config := sarama.NewConfig()

	tlsConfig, err := properties.TLS.ToConfig()
	if err != nil {
		panic(fmt.Errorf("failed to read tls properties for writer: %w", err))
	}

	if tlsConfig != nil {
		config.Net.TLS.Enable = true
		config.Net.TLS.Config = tlsConfig
	}

	config.Producer.Return.Successes = true
	config.Producer.Return.Errors = true
	config.Version = sarama.V2_0_0_0

	config.ClientID = "v8s"

	producer, err := sarama.NewSyncProducer(properties.Addresses, config)
	if err != nil {
		panic(fmt.Errorf("failed to create sync producer: %w", err))
	}

	return &Writer{
		log: logger.Logger().Named("writer").Sugar(),

		producer: producer,
		topic:    properties.ResponseTopic,
	}
}

func (w *Writer) WriteMsg(key, header, value string, msg []byte) {
	w.log.Infof("writing message: %s", string(msg))
	_, _, err := w.producer.SendMessage(&sarama.ProducerMessage{
		Key:   sarama.StringEncoder(key),
		Value: sarama.ByteEncoder(msg),
		Headers: []sarama.RecordHeader{
			{
				Key:   []byte(header),
				Value: []byte(value),
			},
		},
		Topic: w.topic,
	})
	if err != nil {
		w.log.Errorf("failed to write messages: %s", err)
	}
}

func (w *Writer) Shutdown() error {
	return w.producer.Close()
}
